package com.lvhr.rpc.registry;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.collection.ConcurrentHashSet;
import cn.hutool.cron.CronUtil;
import cn.hutool.cron.task.Task;
import cn.hutool.json.JSONUtil;
import com.lvhr.rpc.config.RegistryConfig;
import com.lvhr.rpc.model.ServiceMetaInfo;
import lombok.extern.slf4j.Slf4j;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.params.SetParams;

import java.util.HashSet;
import java.util.List;
import java.util.Objects;
import java.util.Set;
import java.util.stream.Collectors;

@Slf4j
public class RedisRegistry implements Registry {

    private JedisPool jedisPool;

    /**
     * 本机注册的节点 key 集合（用于维护续期）
     */
    private final Set<String> localRegisterNodeKeySet = new HashSet<>();

    /**
     * 注册中心服务缓存（支持多个服务键）
     */
    private final RegistryServiceMultiCache registryServiceMultiCache = new RegistryServiceMultiCache();

    /**
     * 正在监听的 key 集合
     */
    private final Set<String> watchingKeySet = new ConcurrentHashSet<>();

    /**
     * 根节点
     */
    private static final String REDIS_ROOT_PATH = "rpc:";

    /**
     * 心跳间隔时间（秒）
     */
    private static final int HEARTBEAT_INTERVAL = 30;

    @Override
    public void init(RegistryConfig registryConfig) {
        JedisPoolConfig poolConfig = new JedisPoolConfig();
        poolConfig.setMaxTotal(10);
        poolConfig.setMaxIdle(5);
        poolConfig.setMinIdle(2);
        poolConfig.setTestOnBorrow(true);
        poolConfig.setTestOnReturn(true);

        String[] addressParts = registryConfig.getAddress().split(":");
        String host = addressParts[0];
        int port = addressParts.length > 1 ? Integer.parseInt(addressParts[1]) : 6379;

        jedisPool = new JedisPool(poolConfig, host, port,2000
        );

        heartBeat();
    }

    @Override
    public void register(ServiceMetaInfo serviceMetaInfo) throws Exception {
        try (Jedis jedis = jedisPool.getResource()) {
            // 创建存储的键
            String registerKey = REDIS_ROOT_PATH + serviceMetaInfo.getServiceNodeKey();

            // 设置键值对，并设置过期时间
            SetParams setParams = SetParams.setParams()
                    .ex(HEARTBEAT_INTERVAL); // 30秒过期

            jedis.set(registerKey, JSONUtil.toJsonStr(serviceMetaInfo), setParams);

            // 添加到本地缓存
            localRegisterNodeKeySet.add(registerKey);
        }
    }

    @Override
    public void unRegister(ServiceMetaInfo serviceMetaInfo) {
        try (Jedis jedis = jedisPool.getResource()) {
            String registerKey = REDIS_ROOT_PATH + serviceMetaInfo.getServiceNodeKey();
            jedis.del(registerKey);
            localRegisterNodeKeySet.remove(registerKey);
        }
    }

    @Override
    public List<ServiceMetaInfo> serviceDiscovery(String serviceKey) {
        // 优先从缓存获取服务
        List<ServiceMetaInfo> cachedServiceMetaInfoList = registryServiceMultiCache.readCache(serviceKey);
        if (cachedServiceMetaInfoList != null) {
            return cachedServiceMetaInfoList;
        }

        try (Jedis jedis = jedisPool.getResource()) {
            // 前缀搜索
            String searchPattern = REDIS_ROOT_PATH + serviceKey + "/*";

            Set<String> keys = jedis.keys(searchPattern);
            if (CollUtil.isEmpty(keys)) {
                return null;
            }

            // 获取所有匹配的键值
//            List<ServiceMetaInfo> serviceMetaInfoList = keys.stream()
//                    .map(key -> {
//                        // 监听 key 的变化
//                        watch(key);
//                        String value = jedis.get(key);
//                        return JSONUtil.toBean(value, ServiceMetaInfo.class);
//                    })
//                    .collect(Collectors.toList());
            List<ServiceMetaInfo> serviceMetaInfoList = keys.stream()
                    .map(key -> {
                        // 监听 key 的变化
                        watch(key);
                        String value = jedis.get(key);
                        if (value == null) {
                            registryServiceMultiCache.clearCache(key); // 清除无效缓存
                            return null;
                        }
                        return JSONUtil.toBean(value, ServiceMetaInfo.class);
                    })
                    .filter(Objects::nonNull)
                    .collect(Collectors.toList());

            // 写入服务缓存
            registryServiceMultiCache.writeCache(serviceKey, serviceMetaInfoList);
            return serviceMetaInfoList;
        }
    }

    @Override
    public void destroy() {
        log.info("当前节点下线");
        // 下线节点
        try (Jedis jedis = jedisPool.getResource()) {
            for (String key : localRegisterNodeKeySet) {
                jedis.del(key);
            }
        }
        // 释放连接池
        if (jedisPool != null && !jedisPool.isClosed()) {
            jedisPool.close();
        }
    }

    @Override
    public void heartBeat() {
        // 每10秒执行一次心跳检测（续期）
        CronUtil.schedule("*/10 * * * * *", new Task() {
            @Override
            public void execute() {
                try (Jedis jedis = jedisPool.getResource()) {
                    for (String key : localRegisterNodeKeySet) {
                        // 续期
                        jedis.expire(key, HEARTBEAT_INTERVAL);
                        log.debug("续期 key:{}" , key);
                    }
                }
            }
        });

        // 支持秒级别定时任务
        CronUtil.setMatchSecond(true);
        CronUtil.start();
    }

    @Override
    public void watch(String serviceNodeKey) {
        // Redis 原生不支持监听，这里使用定时任务模拟监听
        if (watchingKeySet.add(serviceNodeKey)) {
            // 每10秒检查一次服务变化
            CronUtil.schedule(serviceNodeKey, "*/10 * * * * *", new Task() {
                private String lastValue;

                @Override
                public void execute() {
                    try (Jedis jedis = jedisPool.getResource()) {
                        String currentValue = jedis.get(serviceNodeKey);
                        if (currentValue == null) {
                            // 服务节点已删除
                            registryServiceMultiCache.clearCache(serviceNodeKey);
                            watchingKeySet.remove(serviceNodeKey);
                            CronUtil.remove(serviceNodeKey);
                        } else if (!currentValue.equals(lastValue)) {
                            // 服务节点有变化
                            registryServiceMultiCache.clearCache(serviceNodeKey);
                            lastValue = currentValue;
                        }
                    }
                }
            });
        }
    }
}
